package com.qf.flinketl.source;

import com.qf.flinketl.mapper.SourceMessage;

import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.Properties;

/**
 * @Description: @Author: KyrieFc @Date: 2023/10/22
 */
public class KafkaCommonSource extends RichParallelSourceFunction<SourceMessage> {
  private volatile boolean isRunning = true;
  private static Logger logger = LoggerFactory.getLogger(KafkaCommonSource.class);

  @Override
  public void run(SourceContext<SourceMessage> sourceContext) throws Exception {
    long currentTimeMillis = System.currentTimeMillis();
    // TODO
    Properties properties = new Properties();
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-broker");
    properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "your-consumer-group");
    properties.setProperty(
        ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.setProperty(
        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    KafkaConsumer<SourceMessage, SourceMessage> kafkaConsumer = new KafkaConsumer<>(properties);
    kafkaConsumer.subscribe(Collections.singletonList("your-topic"));
    try {
      while (isRunning) {
        Iterable<ConsumerRecord<SourceMessage, SourceMessage>> records = kafkaConsumer.poll(200);
        for (ConsumerRecord<SourceMessage, SourceMessage> record : records) {
          logger.info("kafka record: {}", record);
          sourceContext.collect(record.value());
        }
      }
    } catch (Exception e) {
      logger.error("kafka consumer error: {}", e.getMessage());
    } finally {
      kafkaConsumer.close();
    }
  }

  @Override
  public void cancel() {
    isRunning = false;
  }
}
